package mr

import (
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"unicode"
)
import "log"
import "net/rpc"
import "hash/fnv"
import "syscall"

//
// Map functions return a slice of KeyValue.
//

type SingleWorker struct {
	Uuid                     int
	Pid                      int    //worker这个线程的pid
	TaskProcessPid           string //worker开的执行任务的子线程的pid
	WorkerName               string
	CurrentTask              string
	CurrentTaskStage         int
	CurrentTaskProgressStage bool //false 未完成，true已完成并提交
	l                        net.Listener
	Mux                      sync.Mutex
	GroupMux                 sync.WaitGroup
	Chan                     chan (string)
}

func port(suffix string) string {
	s := "/var/tmp/824-"
	s += strconv.Itoa(os.Getuid()) + "/"
	os.Mkdir(s, 0777)
	s += "mr"
	s += strconv.Itoa(os.Getpid()) + "-"
	s += suffix
	return s
}

func (wr *SingleWorker) server() {
	rpc.Register(wr)
	rpc.HandleHTTP()
	sockname := wr.WorkerName
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)

	wr.Mux.Lock()
	wr.l = l
	wr.Mux.Unlock()

	if e != nil {
		log.Fatal(sockname+"listen error:", e)
	}
	go http.Serve(l, nil)
}

func (wr *SingleWorker) Ping(args *PingArgs, reply *PingReply) error {

	wr.Mux.Lock()
	defer wr.Mux.Unlock()

	reply.WorkerName = wr.WorkerName
	reply.WorkerCurrentTaskStage = wr.CurrentTaskStage
	reply.WorkerCurrentTaskProgressStage = wr.CurrentTaskProgressStage
	reply.WorkerCurrentTask = wr.CurrentTask

	return nil
}

func MakeWorker() *SingleWorker {
	wr := SingleWorker{
		Uuid:                     -1,
		Pid:                      os.Getpid(),
		WorkerName:               port("worker"),
		TaskProcessPid:           "",
		CurrentTask:              "",
		CurrentTaskStage:         -1,
		CurrentTaskProgressStage: true,
		Chan:                     make(chan (string)),
	}
	return &wr
}

type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//

//func Worker(mapf func(string, string) []KeyValue,
//	reducef func(string, []string) string) {
//
//	// Your worker implementation here.
//
//	// uncomment to send the Example RPC to the master.
//	// CallExample()
//	nReduce := 5
//	var done sync.WaitGroup
//	for i := 0; i < 3; i++ {
//		done.Add(1)
//		go SingleWorker(mapf, reducef, &done, nReduce)
//		//fmt.Fprintf(os.Stderr, "Worker " + strconv.Itoa(i) + " activitated!\n")
//		LogPrintln("Worker " + strconv.Itoa(i) + " activitated!")
//	}
//	done.Wait()
//	return
//
//}

func WriteContent2File(content []KeyValue, fileName string) error {
	ofile, _ := os.Create(fileName)
	for i := 0; i < len(content); i += 1 {
		fmt.Fprintf(ofile, "%v %v\n", content[i].Key, content[i].Value)

	}
	ofile.Close()
	return nil
}

//todo: 明日debug计划，手动把map阶段和reduce阶段分为两部分，明明map阶段的结果都对了，reduce到底是怎么错的搞清楚

func String2kvPair(content string) []KeyValue {
	//strContent := string(content)
	//ff := func(r rune) bool { return !unicode.IsLetter(r) }
	ff := func(r rune) bool { return unicode.IsSpace(r) }
	//todo 出问题
	words := strings.FieldsFunc(content, ff)
	intermediate := []KeyValue{}
	for i := 0; i < len(words)-1; i += 2 {
		tmp := KeyValue{
			Key:   words[i],
			Value: words[i+1],
		}
		intermediate = append(intermediate, tmp)
	}

	return intermediate
}

func WcMapTask(mapf func(string, string) []KeyValue, inputFileAddress string, content string, nReduce int) (map[string]bool, error) {
	kva := mapf(inputFileAddress, string(content))
	tmpInterFileNameSet := make(map[string]bool)
	for i := 0; i < len(kva); i += 1 {
		interFileName := "mr-" + strconv.Itoa(ihash(kva[i].Key)%nReduce)
		tmpInterFileNameSet[interFileName] = true
		ofile, err := os.OpenFile(interFileName, os.O_CREATE|os.O_APPEND|os.O_RDWR, os.ModePerm|os.ModeTemporary)
		if err != nil {
			log.Fatal("Open reduce file: ", err)
		}
		fmt.Fprintf(ofile, "%v %v\n", kva[i].Key, kva[i].Value)
		ofile.Close()
	}

	//返回interFileName list(set实现吧) && error
	return tmpInterFileNameSet, nil
}

func WcReduceTask(reducef func(string, []string) string, content string, taskNumber int) (string, error) {
	finalFileName := "mr-out-" + strconv.Itoa(taskNumber)
	intermediate := String2kvPair(string(content))

	sort.Sort(ByKey(intermediate))

	ofile, err := os.OpenFile(finalFileName, os.O_CREATE|os.O_APPEND|os.O_RDWR, os.ModePerm|os.ModeTemporary)
	if err != nil {
		log.Fatal("Open reduce file: ", err)
	}
	for i := 0; i < len(intermediate); {
		////test-------------------
		////testFlag := false
		//if intermediate[i].Key == "aboard" {
		//	LogPrintln("first location : " + strconv.Itoa(i))
		//	//testFlag = true
		//	for z := 0; z < len(intermediate); z++ {
		//		if intermediate[z].Key == "aboard" {
		//			LogPrintln("location : " + strconv.Itoa(z))
		//		}
		//		//LogPrintln(intermediate[z].Key + " " + intermediate[z].Value)
		//	}
		//}
		////test-------------------
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}

		//if testFlag{
		//	LogPrintln("j is : " + strconv.Itoa(j))
		//	LogPrintln("char is : " + intermediate[j-1].Key)
		//}

		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)

			//if testFlag{
			//	LogPrintln(intermediate[k].Key + intermediate[k].Value)
			//}

		}
		output := reducef(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

		i = j
	}
	ofile.Close()
	return finalFileName, nil
}

func (wr *SingleWorker) ShutdownCurrentTask(args *ShutdownArgs, reply *ShutdownReply) error {
	//终止当前的map or reduce任务
	defer wr.GroupMux.Done()
	numid, _ := strconv.Atoi(wr.TaskProcessPid)
	syscall.Kill(numid, 0)
	wr.TaskProcessPid = ""
	LogPrintln(wr.WorkerName + "'s job was canceled for delaying")
	return nil
}

func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	wr := MakeWorker()

	//defer (*done).Done()

	wr.server()
	LogPrintln(wr.WorkerName + " server online!")

	regReply := CallRegist(wr.WorkerName)

	wr.Mux.Lock()
	wr.Uuid = regReply.WorkerUuId
	wr.Mux.Unlock()

	//uuid := regReply.WorkerUuId
	//todo 需要构建一个worker结构体，放入对应信息，同时上线这个worker对象的服务与master建立连接
	nReduce := 5

	//var subThreadsGroupLock sync.WaitGroup

	//这里非阻塞
	taskReply, ok := CallTask(regReply.WorkerUuId, wr.WorkerName)

	previousTaskType := -1

	for taskReply.WorkType != 2 && ok != false {
		if (taskReply.Flag) == 0 {
			LogPrintln("Worker " + strconv.Itoa(regReply.WorkerUuId) + " receive no task, try again... ")
			taskReply, ok = CallTask(regReply.WorkerUuId, wr.WorkerName)
			continue
		}
		//todo 这里需要改成单个机器可以开多线程(废弃)

		wr.Mux.Lock()
		wr.CurrentTask = taskReply.FileAddress
		wr.CurrentTaskStage = taskReply.WorkType
		wr.CurrentTaskProgressStage = false
		wr.Mux.Unlock()

		file, err := os.Open(taskReply.FileAddress)
		if err != nil {
			log.Fatalf("cannot open %v", taskReply.FileAddress)
		}

		fi, _ := file.Stat() //todo test code获取大小

		content, err := ioutil.ReadAll(file)
		if err != nil {
			log.Fatalf("cannot read %v", taskReply.FileAddress)
		}
		file.Close()

		//workType := taskReply.WorkType
		//fileAddress := taskReply.FileAddress
		//taskNumber := taskReply.TaskNumber
		//go func
		go func(previousTaskType int,
			workType int, fileAddress string,
			taskNumber int, uuid int, pidChan *chan (string), groupLock *sync.WaitGroup) {
			////todo 需要修复一个问题：当map阶段还未结束时，不允许开启reduce任务，由于可能map阶段的某个worker的多线程任务还未完成就开始开reduce的多线程任务，
			////这样导致部分的数据没有被reduce，因此这里需要一个阻塞，需要等到所有的map任务完成后才允许开始接受red uce任务
			////但是：有可能本台机器的map任务全部完成，并不意味着全部机器的map任务都完成了
			//if (previousTaskType != -1 && previousTaskType == workType){
			//	groupLock.Wait()
			//	//等待上一阶段这台机器的全部任务结束，在开启这一阶段的工作
			//}
			////初始or与上一个工作的同种的工作可以直接开启，但是阶段更换需要等到上一阶段的工作完全结束
			//todo :2021年10月12日00:37:40 不对，reduce只有在保证全部map任务完成以后才开始的，不会出现上面的情况，在map全部分配但是还没完成的空档期，应该是重复循环
			//groupLock.Add(1)
			//defer (groupLock).Done() //todo 测试，取消多线程
			//
			groupLock.Add(1)
			defer groupLock.Done()

			(*pidChan) <- strconv.Itoa(os.Getpid())

			if workType == 0 {
				LogPrintln("Worker " + strconv.Itoa(uuid) + " receive a map task " + strconv.Itoa(taskNumber) + " and the file is " + fileAddress)
				tmpInterFileNameSet, _ := WcMapTask(mapf, fileAddress, string(content), nReduce)
				CallTaskFinish(fileAddress, tmpInterFileNameSet, uuid, workType)
			} else {
				LogPrintln("Worker " + strconv.Itoa(uuid) + " receive a reduce task " + strconv.Itoa(taskNumber) + " the file is " + fileAddress + " size is :" + strconv.Itoa((int)(fi.Size())))
				finalFileName, _ := WcReduceTask(reducef, string(content), taskNumber)
				CallTaskFinish(fileAddress, map[string]bool{finalFileName: true}, uuid, workType)
			}

			wr.Mux.Lock()
			wr.CurrentTask = ""
			wr.CurrentTaskStage = -1
			wr.CurrentTaskProgressStage = true
			wr.TaskProcessPid = ""
			wr.Mux.Unlock()

		}(previousTaskType, taskReply.WorkType, taskReply.FileAddress, taskReply.TaskNumber, wr.Uuid, &wr.Chan, &wr.GroupMux)

		wr.TaskProcessPid = <-wr.Chan

		wr.GroupMux.Wait()
		//阻塞，等到map or reduce的线程的结束或者被shoutdown程序终结
		taskReply, ok = CallTask(regReply.WorkerUuId, wr.WorkerName)
	}

	//subThreadsGroupLock.Wait()
	CallExit(wr.Uuid) //todo 这里主动关闭，
	fmt.Fprintf(os.Stderr, "Worker"+strconv.Itoa(regReply.WorkerUuId)+" exit!!! \n")
	wr.l.Close() //rpc服务下线

}

//
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//

func CallTask(workerUuid int, workName string) (*TaskReply, bool) {
	args := TaskArgs{WorkerId: workerUuid, WorkerName: workName}
	reply := TaskReply{}
	ok := call("Master.TaskDistrubution", &args, &reply)
	return &reply, ok
}

func CallRegist(workerName string) *RegistReply {
	args := RegistArgs{}
	args.WorkerName = workerName
	reply := RegistReply{}
	call("Master.Regist", &args, &reply)
	return &reply
}

func CallExit(workerUuid int) *WorkerExitReply {
	args := WorkerExitArgs{
		WorkerId: workerUuid,
	}
	reply := WorkerExitReply{}
	call("Master.Exit", &args, &reply)
	return &reply
}

func CallTaskFinish(Input string, out map[string]bool, workerId int, workType int) *TaskFinishReply {
	args := TaskFinishArgs{
		InputFileName:  Input,
		OutputFileName: out,
		WorkerId:       workerId,
		WorkType:       workType,
	}
	reply := TaskFinishReply{}
	call("Master.TaskFinish", &args, &reply)
	return &reply
}

func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)
	//reply 是rpcname的function的返回值

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//

func WorkerCall(sockname string, rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		//log.Fatal("rpcname is "+ rpcname +" sockname is " + sockname +" WorkerCall() function dialing:" , err)
		errStr := fmt.Sprintf("%s", err)
		LogPrintln("rpcname is " + rpcname + " sockname is " + sockname + " WorkerCall() function dialing: " + errStr)
		return false
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}
	errStr := fmt.Sprintf("%s", err)
	//log.Fatal("rpcname is "+ rpcname +" sockname is " + sockname +" WorkerCall() function call:", err)
	LogPrintln("rpcname is " + rpcname + " sockname is " + sockname + " WorkerCall() function call: " + errStr)
	return false
}

func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		//log.Fatal("call() function dialing:", err)
		errStr := fmt.Sprintf("%s", err)
		LogPrintln("rpcname is " + rpcname + " sockname is " + sockname + " call() function dialing: " + errStr)
		return false
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}
	errStr := fmt.Sprintf("%s", err)
	LogPrintln("rpcname is " + rpcname + " sockname is " + sockname + " call() function call: " + errStr)
	//log.Fatal("rpcname is "+ rpcname +" sockname is " + sockname +" WorkerCall() function call:", err)
	return false
}
